package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.process.IHotArticleProcessor;
import com.heima.behavior.dto.ArticleStreamMessage;
import com.heima.behavior.dto.UpdateArticleMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

/**
 * @author: itheima
 * @create: 2022-04-06 15:00
 */
@EnableBinding(IHotArticleProcessor.class)
public class HotArticleListener {


    /**
     * 监听 hot_article_score_topic 源话题中（用户行为）数据流 形式如下：
     * {"add":1,"articleId":1527136390561685505,"type":1}  用户对某篇文章进行点赞
     * {"add":1,"articleId":1527136390561685505,"type":2}  用户对某篇文章进行评论
     * {"add":1,"articleId":1527136390561685505,"type":3}  用户对某篇文章进行收藏
     * 流式计算逻辑：统计一段时间内（10秒）数据流中包含文章行为数据
     *
     * @param input 数据流中数据，用户行为操作数据
     * @return 将一段时间内统计 对某篇文章所有的行为操作以及增量值 发送到目标话题中 结果如下：
     * {"articleId":1527136390561685505,"like":1,"comment":1,collect:1,view:2}
     */
    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        //1.从监听源话题中获取数据 重新对数据流中进行设置Key（文章ID） Val(操作行为JSON)
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             *
             * @param key 数据流中数据的Key 原始为null
             * @param value 数据流中数据的 原始为：{"add":1,"articleId":1527136390561685505,"type":1}
             * @return 新构建KeyValue对象（新Key 新Val）
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //将行为操作转为Java对象 UpdateArticleMessage
                UpdateArticleMessage message = JSON.parseObject(value, UpdateArticleMessage.class);
                return new KeyValue<>(message.getArticleId().toString(), value);
            }
        });
        //2.对数据进行分组：根据文章ID进行分组  得到数据流：Key:文章ID  val:用户操作行为JSON
        KGroupedStream<String, String> groupedStream = map.groupByKey();

        //3.设置时间窗口：统计10秒中以内某篇文章所有操作行为
        TimeWindowedKStream<String, String> timeWindowedKStream = groupedStream.windowedBy(TimeWindows.of(Duration.ofSeconds(10)));

        //4.设置聚合：对某篇文章所有操作进行聚合（得到所有操作）

        //4.1 构建初始化聚合结果对象
        Initializer initializer = () -> {
            //第一条记录消息进入后 聚合结果为空
            return null;
        };
        //4.2 构建聚合结果:对10秒内某篇文章进行聚合
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {

            /**
             * 返回新的聚合结果
             * @param key 数据流中数据Key： 文章ID
             * @param value 数据流中数据Val： 行为操作JSON  {"add":1,"articleId":1527136390561685505,"type":2}
             * @param aggregate 上次聚合结果
             * @return 某篇文章本次聚合结果（上次+本次）
             * {"articleId":1527136390561685505,"like":1,"comment":1,collect:1,view:2}
             */
            @Override
            public String apply(String key, String value, String aggregate) {
                //4.2.1 声明某篇文章所有聚合结果对象 ArticleStreamMessage
                ArticleStreamMessage result = new ArticleStreamMessage();

                //4.2.2 判断上次聚合是否有值
                if (StringUtils.isBlank(aggregate)) {
                    result.setArticleId(Long.parseLong(key));
                    result.setView(0);
                    result.setCollect(0);
                    result.setComment(0);
                    result.setLike(0);
                } else {
                    //说明上次有聚合结果
                    result = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                //4.2.3 根据本次操作行为更新聚合结果
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //得到本次文章操作增量值
                Integer add = updateArticleMessage.getAdd();
                switch (updateArticleMessage.getType()) {
                    case 0:
                        //阅读
                        result.setView(result.getView() + add);
                        break;
                    case 1:
                        //点赞
                        result.setLike(result.getLike() + add);
                        break;
                    case 2:
                        //评论
                        result.setComment(result.getComment() + add);
                        break;
                    case 3:
                        //收藏
                        result.setCollect(result.getCollect() + add);
                        break;
                }
                return JSON.toJSONString(result);
            }
        };

        //4.3 设置聚合
        KTable<Windowed<String>, String> aggregate = timeWindowedKStream.aggregate(initializer, aggregator);

        //5.将实时计算结果发送到目标话题（处理Key val）
        KStream<String, String> kStream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return kStream;
    }
}
